package com.qf.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.qf.config.DelayedConfig;
import com.qf.constant.CacheConstants;
import com.qf.constant.RabbitMQConstants;
import com.qf.model.StandardReport;
import com.qf.service.api.CacheService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.Map;

/**
 * @author 何星宇
 * @version V1.0
 * @Date 2021/11/24 20:20
 */
@Component
@Slf4j
public class ReportListener {
    @Autowired
    private CacheService cacheService;
    @Autowired
    private RestTemplate restTemplate;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final String[] waitTimes = {"0","5000","10000","30000","60000"};

    //report监听器
    @RabbitListener(queues = RabbitMQConstants.TOPIC_PUSH_SMS_REPORT)
    public void reportListener(StandardReport standardReport , Channel channel, Message message) throws Exception {
        //1.接收report队列中的消息对象，根据消息对象到redis中找到client对象
        Map<String,String> client = cacheService.hMGet(CacheConstants.CACHE_PREFIX_CLIENT + standardReport.getClientID());
        if (client==null){
            log.info("【接口模块-推送业务】 获取client对象为空！ standardReport={}",standardReport);
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        //2.从client对象中找到客户项目的回调地址
        String url = client.get("receivestatusurl");
        //获取客户是否需要推送
        String status = client.get("isreturnstatus");
        if ("0".equals(status)){
            //记录日志
            log.info("【接口模块-推送业务】 客户不需要推送，执行完成！ standardReport={}",standardReport);
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }


        //调用推送方法
        String result = sendClient(url,standardReport);


        //4.如果接收到返回的success，证明客户接收到，不继续发送
        if ("SUCCESS".equalsIgnoreCase(result)) {
            //记录日志
            log.info("【接口模块-推送业务】 推送成功，短信发送完成！");
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }else {
            //如果没有收到success，将report对象发送到mq的延时队列
            sendDealyde(standardReport);
            log.info("【接口模块-推送业务】 没有收到success，report对象发送到mq的延时队列");
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
    }

    //dealyed延时队列监听器
    @RabbitListener(queues = DelayedConfig.DELAYED_QUEUE)
    public void dealydeListener(StandardReport standardReport , Channel channel, Message message) throws Exception {

        //获取回调地址
        Map<String,String> client = cacheService.hMGet(CacheConstants.CACHE_PREFIX_CLIENT + standardReport.getClientID());
        String url = client.get("receivestatusurl");

        //调用推送方法
        String result = sendClient(url,standardReport);

        //2.如果客户返回success，那么不继续发送，发送停止
        if ("SUCCESS".equalsIgnoreCase(result)){
            //记录日志
            log.info("【接口模块-推送业务】 推送成功，短信发送完成！");
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        //3.如果客户没有返回success，那么判断发送次数，如果不到5次，那么report对象重新放入mq的延时队列等待发送
        if (standardReport.getSendCount() < 5){
            //将report对象发送到mq的延时队列
            log.info("【接口模块-推送业务】 没有收到success，report对象发送到mq的延时队列");
            sendDealyde(standardReport);
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        //4.如果发送次数达到了5次，发送停止
        return;
    }


    //推送方法
    public String sendClient(String url,StandardReport standardReport) throws Exception {
        //发送次数+1
        standardReport.setSendCount(standardReport.getSendCount()+1);
        //将对象转换为json
        String report = objectMapper.writeValueAsString(standardReport);

        //封装请求头对象
        HttpHeaders head = new HttpHeaders();
        head.setContentType(MediaType.APPLICATION_JSON);

        //封装请求体对象
        HttpEntity<String> entity = new HttpEntity<>(report,head);

        //发送
        return restTemplate.postForObject(url, entity, String.class);
    }


    //发送到延迟队列
    public void sendDealyde(StandardReport standardReport){
        rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "report", standardReport, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration(waitTimes[standardReport.getSendCount()]);
                return message;
            }
        });
    }
}
